package com.k8.rpc.exchange.impl;

import com.k8.common.aware.BeanFactoryAware;
import com.k8.common.factroy.BeanFactory;
import com.k8.common.factroy.loader.ExtensionLoader;
import com.k8.rpc.exchange.Exchange;
import com.k8.rpc.exchange.ExchangeBuilder;
import com.k8.rpc.exchange.ExchangeManager;
import com.k8.rpc.url.URL;

import javax.annotation.Resource;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import static com.k8.common.constants.CommonKeyConstants.CONSUMER;

/**
 * @Author: k8
 * @CreateTime: 2025-04-07
 * @Version: 1.0
 */
public class DefaultExchangeManager implements ExchangeManager, BeanFactoryAware {
    private BeanFactory beanFactory;
    private final ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    private final Object lock = new Object();
    private volatile boolean isAlive = true;
    private final ReadWriteLock aliveLock = new ReentrantReadWriteLock(true);

    /**
     * 避免相同的应用实例建立多个连接池
     *
     * @param path
     * @param subUrl
     * @return
     */
    @Override
    public Exchange getOrRegistryExchange(String path, URL subUrl) {
        if (!isAlive) throw new IllegalStateException("exchangeManger is shutDown");
        Lock readLock = aliveLock.readLock();
        try {
            readLock.lock();
            if (!isAlive) throw new IllegalStateException("exchangeManger is shutDown");
        } finally {
            readLock.unlock();
        }
        String side = subUrl.getSide();
        path += side;
        Exchange exchange = exchangeMap.get(path);
        if (exchange == null) {
            synchronized (lock) {
                exchange = exchangeMap.get(path);
                if (exchange == null) {
                    exchange = crateExchange(subUrl, side);
                    exchangeMap.put(path, exchange);
                }
            }
        }
        return exchange;
    }

    @Override
    public void shutDown() {
        if (!isAlive) return;
        Lock writeLock = aliveLock.writeLock();
        try {
            writeLock.lock();
            if (!isAlive) return;
            exchangeMap.forEach((key, exchange) -> {
                exchange.shutDown();
            });
            isAlive = false;
        } finally {
            writeLock.unlock();
        }
    }

    private Exchange crateExchange(URL suburl, String side) {
        ExtensionLoader<ExchangeBuilder> exchangeBuilderExtensionLoader = beanFactory.getExtensionDirector().getExtensionLoader(ExchangeBuilder.class);
        ExchangeBuilder exchangeBuilder = exchangeBuilderExtensionLoader.getExtension(suburl.getExchange());
        if (exchangeBuilder == null) {
            exchangeBuilder = exchangeBuilderExtensionLoader.getExtensions().get(0);
            if (exchangeBuilder == null) throw new IllegalStateException("The exchangeBuilder is null.");
        }
        Exchange exchange = exchangeBuilder.build();
        if (CONSUMER.equals(side)) {
            exchange.preConnect(suburl);
        }
        return exchange;
    }

    @Override
    public void beanFactoryAware(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }
}
